package df

import org.apache.spark.sql.{DataFrame, SparkSession}

object E5_kafka {
  // 创建 SparkSession
  val spark = SparkSession.builder()
    .appName("HelloStructuredStreaming")
    .master("local[*]")
    .getOrCreate()

  // 导入隐式转换

  import spark.implicits._

  def main(args: Array[String]): Unit = {
    out_kafka
    read_kafka
  }

  def read_kafka = {
    // 创建一个流式DataFrame，这里从Kafka中读取数据
    val lines: DataFrame = spark.read // 使用 read 方法,而不是 readStream 方法
      .format("kafka") // 设置 kafka 数据源
      .option("kafka.bootstrap.servers", "localhost:9092")
      .option("subscribe", "spark-topic") // 也可以订阅多个主题 "topic1,topic2"
      .option("startingOffsets", "earliest")
      .option("endingOffsets", "latest")
      .load

    // 数据处理
    val wordCounts = lines
      .select("value").as[String] // 将每条消息的 value 取出（不需要key）
      .map {
        value => {
          val arr = value.split(",")
          (arr(0), arr(1))
        }
      }
      .toDF("Word", "Count") // 添加列名

    // 把结果打印到控制台
    wordCounts.show()

    //关闭 Spark
    spark.stop()
  }

  def out_kafka = {
    // 创建一个静态DataFrame，这里通过数组创建
    val wordCount: DataFrame = spark.sparkContext
      .parallelize(Array("hello hangge", "hi hangge"))
      .toDF("word")
      .flatMap(_.getString(0).split("\\W+")) // 使用flatMap拆分和展平
      .groupBy("value")
      .count()
      .map(row => row.getString(0) + "," + row.getLong(1))
      .toDF("value") // 写入数据时候, 必须有一列 "value"

    // 启动查询, 把结果输出至kafka
    val query = wordCount.write
      .format("kafka") // 输出至kafka
      .option("kafka.bootstrap.servers", "localhost:9092") // kafka 配置
      .option("topic", "spark-topic") // kafka 主题
      .save()

    //关闭 Spark
    //    spark.stop()
  }

}
